package com.qingyan.flink.core.sink;

import com.qingyan.flink.core.Table;
import com.qingyan.flink.core.properties.FlinkDataSourceProperties;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.AbstractCatalog;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.logical.LogicalType;
import java.util.List;
import java.util.Map;

/**
 * @author csp 2023-12-27
 */
public class JdbcCatalogSinkBuilder extends AbstractSinkBuilder{

    protected AbstractCatalog catalog;

    public JdbcCatalogSinkBuilder(FlinkDataSourceProperties sinkProperties, Map<String, String> sinkConfig) {
        super(sinkProperties, sinkConfig);
    }

    public JdbcCatalogSinkBuilder() {
    }

    @Override
    public String getSinkSchemaName(Table table) {
        return null;
    }

    @Override
    public String getSinkTableName(Table table) {
        return null;
    }

    @Override
    public void addSink(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, DataStream<RowData> rowDataDataStream, Table table, List<String> columnNameList, List<LogicalType> columnTypeList) {
        String catalogName = catalog.getName();
        tableEnv.useCatalog(catalogName);
        tableEnv.useDatabase(table.getSchema());
        org.apache.flink.table.api.Table t = tableEnv.fromDataStream(rowDataDataStream);
        t.insertInto(table.getName(), true);
    }



    @Override
    public String getHandle() {
        return null;
    }

    @Override
    public SinkBuilder create(FlinkDataSourceProperties properties, Map<String, String> sourceConfig) {
        return new JdbcCatalogSinkBuilder(sinkProperties, sinkConfig);
    }



}
